package cn.itcast.mr.cases;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

/**
 * Author itcast
 * Date 2021/10/17 17:13
 * Desc TODO
 */
public class CommonFriendStepOne {
    public static class CommonFriendsMapper extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (StringUtils.isNotEmpty(line)) {
                String[] userFriends = line.split(":");
                String friends = userFriends[1];
                String user = userFriends[0];
                if (friends != null){
                    String[] friendArr = friends.split(",");
                    for (String friend : friendArr) {
                        context.write(new Text(friend),new Text(user));
                    }
                }
            }
        }
        public static class CommonFriendsReducer extends Reducer<Text,Text,Text,Text>{
            @Override
            protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
                StringBuffer buffer = new StringBuffer();
                for (Text user : values) {
                    buffer.append(user);
                    buffer.append("-");
                }
                buffer.deleteCharAt(buffer.length()-1);
                context.write(new Text(buffer.toString()),key);
            }
        }

        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            //创建 Job 实例
            Job job = Job.getInstance();
            //job设置输入的格式化类型
            job.setInputFormatClass(TextInputFormat.class);
            //设置输入的文件路径
            TextInputFormat.addInputPath(job, new Path("D:\\project\\mr\\data\\friends.txt"));
            //job设置map的实现类
            job.setMapperClass(CommonFriendsMapper.class);
            //job设置map的输出的key类型
            job.setMapOutputKeyClass(Text.class);
            //job设置map的输出的value类型
            job.setMapOutputValueClass(Text.class);

            //job设置shuffle，设置自定义了分区
            //job设置shuffle，设置局部聚合 combiner

            //job设置reduce的实现类
            job.setReducerClass(CommonFriendsReducer.class);
            //job设置reduce的输出key
            job.setOutputKeyClass(Text.class);
            //job设置reduce的输出value
            job.setOutputValueClass(Text.class);

            //job设置reducer输出到文件的输出类型
            job.setOutputFormatClass(TextOutputFormat.class);
            //job设置输出的路径
            TextOutputFormat.setOutputPath(job, new Path("D:\\project\\mr\\data\\friends_result01"));

            //job等待执行
            boolean flag = job.waitForCompletion(true);
            //返回程序执行的结果，如果成功返回为0，否则1
            System.exit(flag ? 0 : 1);
        }
    }
}
